package com.spark.sql

import org.apache.spark.sql.SparkSession

object DataFrameAPI {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder().appName("Rdd2DF").master("local[1]").getOrCreate()

    // load data as rdd
    val rdd = spark.sparkContext.textFile("file:///Users/username/workspace_code/learn/spark-learn/datasets/infos2.txt")
//    rdd.foreach(println)

    val colRdd = rdd.map(_.split("\\|")).map(line => Info(line(0).toInt, line(1), line(2).toInt, line(3).toInt, line(4)))

    // rdd to df
    import spark.implicits._
    val infoDF = colRdd.toDF()
    // 默认显示前20行
//    infoDF.show()
    // 第一个参数指定显示多少行，第二个指定是否要截断显示（当字段长度较长时，默认情况下会自动截断）
//    infoDF.show(30, false)

//    infoDF.take(10).foreach(println)   // 打印前10行

//    infoDF.first()   // 显示第一行
//    infoDF.head(10)   // 显示前n行

    infoDF.select("name", "email").show(5)

    // 条件过滤
    infoDF.filter("name='' or name='NULL'").show()

    // 字段名过滤， substr为spark自带函数，可以通过spark.sql("show functions").show() 查询
    infoDF.filter("substr(name, 0, 1)='张'").show()

    // 查看可用函数
//    spark.sql("show functions").show(1000)

    // 先按name降序排序，再按id升序排
    infoDF.sort(infoDF("name").desc, infoDF("id").asc).show()
    infoDF.sort(infoDF.col("name").desc, infoDF("id").asc).show()   // 两个指定字段名方式都可以

    // 字段重命名
    infoDF.select(infoDF("name").as("stu_name")).show()

    // 多个DF的join操作
    val infoDF2 = colRdd.toDF()

    // 1. 相等是三个=号， 2.  可以outer，inner，right，left等多种join方式
    infoDF.join(infoDF2, infoDF.col("id") === infoDF2.col("id"), "inner").show()
    spark.stop()

  }
  case class Info(id: Int, name: String, age: Int, height: Int, email: String)
}
